package exlocalsubpub

import (
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/gobase/factory"
	"gitee.com/ymofen/gobase/subpub"
	"io"
	"sync/atomic"
)

// 将数据分流给NtripClient
type PubExLocalSession struct {
	closedflag int32
	notify     subpub.ISubPub
	sessionid  string

	conf gobase.StrMap

	copybufflag int8

	operaRef *gobase.ObjectInnerRef
	opera    *ExLocalNode
}

func CreatePubExLocalSession(args ...interface{}) (rval interface{}, err error) {
	plg := &PubExLocalSession{
		sessionid: args[0].(string),
		conf:      args[1].(gobase.StrMap),
		notify:    args[2].(subpub.ISubPub),
	}
	return plg, nil
}

func (this *PubExLocalSession) Pub(topic string, max int, args ...interface{}) int {
	//ex := this.notify
	//if ex == nil {
	//	return -1
	//}
	//
	//if len(args) == 0 {
	//	ex.Pub(this.sessionid, 0, subpub.ACTION_ERROR, fmt.Errorf("没有要Pub的数据"))
	//	return -1
	//}

	opera := this.opera
	if opera == nil {
		return -1
	}

	return opera.Pub(topic, max, args...)
}

func (this *PubExLocalSession) Start() {
	connstr := this.conf.URLFormEncode()
	this.operaRef = DefaultExGroup.CheckGetRef(connstr)
	this.opera = this.operaRef.GetObject().(*ExLocalNode)
	this.opera.ex.Sub(this.sessionid, "evt", func(id, topic string, args ...interface{}) (ok bool) {
		return true
	})
	this.opera.checkStart()
}

func (this *PubExLocalSession) Close() error {
	if !atomic.CompareAndSwapInt32(&this.closedflag, 0, 1) {
		return io.ErrClosedPipe
	}
	if this.operaRef != nil {
		this.opera.ex.Unsub(this.sessionid, "evt")
		this.opera = nil

		DefaultExGroup.ReleaseRef(this.operaRef)
		this.operaRef = nil
	}
	this.notify.Pub(this.sessionid, 0, subpub.ACTION_DISCONNECTED)
	this.notify = nil
	return nil
}

func init() {
	factory.RegisterCreateFunc("pub.exlocal", CreatePubExLocalSession)
}
